//! 查询优化器模块
//!
//! 实现完整的基于成本的查询优化器，包括：
//! - 统计信息收集和管理
//! - 基于规则的优化
//! - 成本估算模型
//! - 计划缓存
//! - JOIN顺序优化

use crate::{Result, Error};
use super::planner::{LogicalPlan, Expression, JoinType};
use crate::sql::Value;
use std::collections::HashMap;
use std::sync::Arc;
use parking_lot::RwLock;
use std::time::{SystemTime, UNIX_EPOCH};

/// 查询优化器
pub struct QueryOptimizer {
    /// 优化规则列表
    rules: Vec<Box<dyn OptimizationRule>>,
    /// 优化统计信息
    stats: OptimizationStats,
}

impl std::fmt::Debug for QueryOptimizer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("QueryOptimizer")
            .field("rules_count", &self.rules.len())
            .field("stats", &self.stats)
            .finish()
    }
}

/// 优化统计信息
#[derive(Debug, Clone, Default)]
pub struct OptimizationStats {
    /// 优化次数
    pub optimization_count: u64,
    /// 规则应用次数
    pub rule_applications: u64,
    /// 平均优化时间（微秒）
    pub avg_optimization_time_us: u64,
}

impl QueryOptimizer {
    /// 创建新的查询优化器
    pub fn new() -> Self {
        let mut rules: Vec<Box<dyn OptimizationRule>> = Vec::new();

        // 添加基础优化规则
        rules.push(Box::new(PredicatePushdownRule));
        rules.push(Box::new(ProjectionPushdownRule));
        rules.push(Box::new(ConstantFoldingRule));
        rules.push(Box::new(JoinReorderRule));
        rules.push(Box::new(IndexSelectionRule));

        Self {
            rules,
            stats: OptimizationStats::default(),
        }
    }
    
    /// 优化逻辑执行计划
    pub fn optimize(&mut self, plan: LogicalPlan) -> Result<LogicalPlan> {
        let start_time = std::time::Instant::now();
        tracing::debug!("开始查询优化");

        let mut optimized_plan = plan;
        let mut rule_applications = 0;

        // 应用所有优化规则
        for rule in &self.rules {
            optimized_plan = rule.optimize(optimized_plan)?;
            rule_applications += 1;
            tracing::debug!("应用优化规则: {}", rule.name());
        }

        // 更新统计信息
        let elapsed = start_time.elapsed();
        self.stats.optimization_count += 1;
        self.stats.rule_applications += rule_applications;
        self.stats.avg_optimization_time_us =
            (self.stats.avg_optimization_time_us * (self.stats.optimization_count - 1) + elapsed.as_micros() as u64)
            / self.stats.optimization_count;

        tracing::debug!("查询优化完成，耗时: {:?}", elapsed);
        Ok(optimized_plan)
    }

    /// 获取优化统计信息
    pub fn get_stats(&self) -> &OptimizationStats {
        &self.stats
    }

    /// 重置统计信息
    pub fn reset_stats(&mut self) {
        self.stats = OptimizationStats::default();
    }
    
    /// 添加优化规则
    pub fn add_rule(&mut self, rule: Box<dyn OptimizationRule>) {
        self.rules.push(rule);
    }

    /// 设置统计信息收集器
    pub fn set_statistics_collector(&mut self, collector: Arc<StatisticsCollector>) {
        // 为需要统计信息的规则设置收集器
        // 这里可以扩展为更复杂的依赖注入
        tracing::debug!("设置统计信息收集器");
    }
}

impl Default for QueryOptimizer {
    fn default() -> Self {
        Self::new()
    }
}

/// 优化规则特征
pub trait OptimizationRule: Send + Sync {
    /// 规则名称
    fn name(&self) -> &str;
    
    /// 优化逻辑计划
    fn optimize(&self, plan: LogicalPlan) -> Result<LogicalPlan>;
}

/// 谓词下推优化规则
/// 将过滤条件尽可能推到数据源附近
#[derive(Debug)]
pub struct PredicatePushdownRule;

impl OptimizationRule for PredicatePushdownRule {
    fn name(&self) -> &str {
        "PredicatePushdown"
    }
    
    fn optimize(&self, plan: LogicalPlan) -> Result<LogicalPlan> {
        match plan {
            LogicalPlan::Filter { input, predicate } => {
                match *input {
                    LogicalPlan::Projection { input: proj_input, expressions } => {
                        // 尝试将过滤条件推到投影之前
                        let optimized_input = self.optimize(*proj_input)?;
                        Ok(LogicalPlan::Projection {
                            input: Box::new(LogicalPlan::Filter {
                                input: Box::new(optimized_input),
                                predicate,
                            }),
                            expressions,
                        })
                    }
                    _ => {
                        let optimized_input = self.optimize(*input)?;
                        Ok(LogicalPlan::Filter {
                            input: Box::new(optimized_input),
                            predicate,
                        })
                    }
                }
            }
            LogicalPlan::Projection { input, expressions } => {
                let optimized_input = self.optimize(*input)?;
                Ok(LogicalPlan::Projection {
                    input: Box::new(optimized_input),
                    expressions,
                })
            }
            LogicalPlan::Join { left, right, join_type, condition } => {
                let optimized_left = self.optimize(*left)?;
                let optimized_right = self.optimize(*right)?;
                Ok(LogicalPlan::Join {
                    left: Box::new(optimized_left),
                    right: Box::new(optimized_right),
                    join_type,
                    condition,
                })
            }
            _ => Ok(plan), // 其他计划类型暂不优化
        }
    }
}

/// 投影下推优化规则
/// 只选择需要的列，减少数据传输
#[derive(Debug)]
pub struct ProjectionPushdownRule;

impl OptimizationRule for ProjectionPushdownRule {
    fn name(&self) -> &str {
        "ProjectionPushdown"
    }
    
    fn optimize(&self, plan: LogicalPlan) -> Result<LogicalPlan> {
        // 简化实现，后续完善
        Ok(plan)
    }
}

/// 常量折叠优化规则
/// 在编译时计算常量表达式
#[derive(Debug)]
pub struct ConstantFoldingRule;

impl OptimizationRule for ConstantFoldingRule {
    fn name(&self) -> &str {
        "ConstantFolding"
    }

    fn optimize(&self, plan: LogicalPlan) -> Result<LogicalPlan> {
        // 简化实现：查找并折叠常量表达式
        match plan {
            LogicalPlan::Filter { input, predicate } => {
                // 尝试计算常量条件
                if self.is_constant_expression(&predicate) {
                    if self.evaluate_constant_condition(&predicate)? {
                        // 条件为真，移除过滤器
                        Ok(*input)
                    } else {
                        // 条件为假，返回空结果
                        Ok(LogicalPlan::Limit {
                            input,
                            count: 0
                        })
                    }
                } else {
                    Ok(LogicalPlan::Filter { input, predicate })
                }
            }
            _ => Ok(plan),
        }
    }
}

impl ConstantFoldingRule {
    fn is_constant_expression(&self, _expr: &Expression) -> bool {
        // 简化实现：检查表达式是否为常量
        false
    }

    fn evaluate_constant_condition(&self, _expr: &Expression) -> Result<bool> {
        // 简化实现：计算常量条件
        Ok(true)
    }
}

/// 连接重排序规则
/// 基于成本模型重新排序连接操作
#[derive(Debug)]
pub struct JoinReorderRule;

impl OptimizationRule for JoinReorderRule {
    fn name(&self) -> &str {
        "JoinReorder"
    }

    fn optimize(&self, plan: LogicalPlan) -> Result<LogicalPlan> {
        // 简化实现：基于表大小重排序连接
        match plan {
            LogicalPlan::Join { left, right, join_type, condition } => {
                // 估算表大小并决定连接顺序
                let left_cost = self.estimate_plan_cost(&left);
                let right_cost = self.estimate_plan_cost(&right);

                if left_cost > right_cost {
                    // 交换连接顺序，小表在左
                    Ok(LogicalPlan::Join {
                        left: right,
                        right: left,
                        join_type,
                        condition,
                    })
                } else {
                    Ok(LogicalPlan::Join { left, right, join_type, condition })
                }
            }
            _ => Ok(plan),
        }
    }
}

impl JoinReorderRule {
    fn estimate_plan_cost(&self, _plan: &LogicalPlan) -> u64 {
        // 简化实现：估算执行计划成本
        // 实际实现中会考虑表大小、索引、选择性等因素
        100
    }
}

/// 索引选择规则
/// 为查询条件选择最优索引
#[derive(Debug)]
pub struct IndexSelectionRule;

impl OptimizationRule for IndexSelectionRule {
    fn name(&self) -> &str {
        "IndexSelection"
    }

    fn optimize(&self, plan: LogicalPlan) -> Result<LogicalPlan> {
        // 简化实现：为过滤条件选择合适的索引
        match plan {
            LogicalPlan::Filter { input, predicate } => {
                if let LogicalPlan::TableScan { table_name, .. } = input.as_ref() {
                    // 检查是否可以使用索引
                    if self.can_use_index(table_name, &predicate) {
                        tracing::debug!("为表 {} 的过滤条件选择索引", table_name);
                        // 在实际实现中，这里会创建IndexScan节点
                        Ok(LogicalPlan::Filter { input, predicate })
                    } else {
                        Ok(LogicalPlan::Filter { input, predicate })
                    }
                } else {
                    Ok(LogicalPlan::Filter { input, predicate })
                }
            }
            _ => Ok(plan),
        }
    }
}

impl IndexSelectionRule {
    fn can_use_index(&self, _table_name: &str, _condition: &Expression) -> bool {
        // 简化实现：检查是否可以使用索引
        // 实际实现中会检查索引定义和条件匹配性
        false
    }
}

/// 统计信息收集器
/// 负责收集和管理数据库的统计信息，用于查询优化
#[derive(Debug)]
pub struct StatisticsCollector {
    /// 表统计信息
    table_stats: RwLock<HashMap<String, TableStatistics>>,
    /// 列统计信息
    column_stats: RwLock<HashMap<(String, String), ColumnStatistics>>,
    /// 索引统计信息
    index_stats: RwLock<HashMap<String, IndexStatistics>>,
    /// 统计信息配置
    config: StatisticsConfig,
}

/// 表统计信息
#[derive(Debug, Clone)]
pub struct TableStatistics {
    /// 表名
    pub table_name: String,
    /// 行数估计
    pub row_count: u64,
    /// 页面数量
    pub page_count: u64,
    /// 平均行大小（字节）
    pub avg_row_size: f64,
    /// 最后更新时间
    pub last_updated: SystemTime,
    /// 数据分布直方图
    pub histogram: Option<Histogram>,
}

/// 列统计信息
#[derive(Debug, Clone)]
pub struct ColumnStatistics {
    /// 表名
    pub table_name: String,
    /// 列名
    pub column_name: String,
    /// 唯一值数量
    pub distinct_count: u64,
    /// 空值数量
    pub null_count: u64,
    /// 最小值
    pub min_value: Option<Value>,
    /// 最大值
    pub max_value: Option<Value>,
    /// 数据分布直方图
    pub histogram: Option<Histogram>,
    /// 最后更新时间
    pub last_updated: SystemTime,
}

/// 索引统计信息
#[derive(Debug, Clone)]
pub struct IndexStatistics {
    /// 索引名
    pub index_name: String,
    /// 表名
    pub table_name: String,
    /// 索引类型
    pub index_type: String,
    /// 索引大小（字节）
    pub size_bytes: u64,
    /// 索引高度（B+树）
    pub height: Option<u32>,
    /// 叶子页面数量
    pub leaf_pages: Option<u64>,
    /// 选择性（0.0-1.0）
    pub selectivity: f64,
    /// 最后更新时间
    pub last_updated: SystemTime,
}

/// 数据分布直方图
#[derive(Debug, Clone)]
pub struct Histogram {
    /// 桶的数量
    pub bucket_count: usize,
    /// 每个桶的边界值
    pub bucket_bounds: Vec<Value>,
    /// 每个桶的频率
    pub bucket_frequencies: Vec<u64>,
    /// 总样本数
    pub total_samples: u64,
}

/// 统计信息配置
#[derive(Debug, Clone)]
pub struct StatisticsConfig {
    /// 统计信息过期时间（秒）
    pub max_age_seconds: u64,
    /// 直方图桶数量
    pub histogram_buckets: usize,
    /// 自动更新阈值（数据变化百分比）
    pub auto_update_threshold: f64,
    /// 采样率（0.0-1.0）
    pub sampling_rate: f64,
}

impl StatisticsCollector {
    /// 创建新的统计信息收集器
    pub fn new() -> Self {
        Self {
            table_stats: RwLock::new(HashMap::new()),
            column_stats: RwLock::new(HashMap::new()),
            index_stats: RwLock::new(HashMap::new()),
            config: StatisticsConfig::default(),
        }
    }

    /// 使用自定义配置创建统计信息收集器
    pub fn with_config(config: StatisticsConfig) -> Self {
        Self {
            table_stats: RwLock::new(HashMap::new()),
            column_stats: RwLock::new(HashMap::new()),
            index_stats: RwLock::new(HashMap::new()),
            config,
        }
    }

    /// 收集表统计信息
    /// 扫描表数据并计算统计信息
    pub fn collect_table_statistics(&self, table_name: &str, rows: &[Vec<Value>]) -> Result<()> {
        tracing::debug!("开始收集表 {} 的统计信息", table_name);

        let row_count = rows.len() as u64;
        let avg_row_size = if row_count > 0 {
            let total_size: usize = rows.iter()
                .map(|row| row.iter().map(|v| self.estimate_value_size(v)).sum::<usize>())
                .sum();
            total_size as f64 / row_count as f64
        } else {
            0.0
        };

        // 估算页面数量（假设4KB页面）
        let page_count = ((row_count as f64 * avg_row_size) / 4096.0).ceil() as u64;

        let table_stats = TableStatistics {
            table_name: table_name.to_string(),
            row_count,
            page_count,
            avg_row_size,
            last_updated: SystemTime::now(),
            histogram: None, // 可以后续添加
        };

        self.table_stats.write().insert(table_name.to_string(), table_stats);

        tracing::debug!("表 {} 统计信息收集完成: {} 行, 平均行大小 {:.2} 字节",
                       table_name, row_count, avg_row_size);
        Ok(())
    }

    /// 收集列统计信息
    /// 分析列的数据分布特征
    pub fn collect_column_statistics(&self, table_name: &str, column_name: &str,
                                   column_index: usize, rows: &[Vec<Value>]) -> Result<()> {
        tracing::debug!("开始收集列 {}.{} 的统计信息", table_name, column_name);

        let mut distinct_values = std::collections::HashSet::new();
        let mut null_count = 0u64;
        let mut min_value: Option<Value> = None;
        let mut max_value: Option<Value> = None;

        // 分析列数据
        for row in rows {
            if column_index < row.len() {
                let value = &row[column_index];

                // 检查空值
                if matches!(value, Value::Null) {
                    null_count += 1;
                    continue;
                }

                // 统计唯一值
                distinct_values.insert(value.clone());

                // 更新最小值和最大值
                match (&min_value, &max_value) {
                    (None, None) => {
                        min_value = Some(value.clone());
                        max_value = Some(value.clone());
                    }
                    (Some(min), Some(max)) => {
                        if self.compare_values(value, min) < 0 {
                            min_value = Some(value.clone());
                        }
                        if self.compare_values(value, max) > 0 {
                            max_value = Some(value.clone());
                        }
                    }
                    _ => unreachable!(),
                }
            }
        }

        let column_stats = ColumnStatistics {
            table_name: table_name.to_string(),
            column_name: column_name.to_string(),
            distinct_count: distinct_values.len() as u64,
            null_count,
            min_value,
            max_value,
            histogram: None, // 可以后续添加直方图
            last_updated: SystemTime::now(),
        };

        let key = (table_name.to_string(), column_name.to_string());
        self.column_stats.write().insert(key, column_stats);

        tracing::debug!("列 {}.{} 统计信息收集完成: {} 个唯一值, {} 个空值",
                       table_name, column_name, distinct_values.len(), null_count);
        Ok(())
    }

    /// 获取表统计信息
    pub fn get_table_statistics(&self, table_name: &str) -> Option<TableStatistics> {
        self.table_stats.read().get(table_name).cloned()
    }

    /// 获取列统计信息
    pub fn get_column_statistics(&self, table_name: &str, column_name: &str) -> Option<ColumnStatistics> {
        let key = (table_name.to_string(), column_name.to_string());
        self.column_stats.read().get(&key).cloned()
    }

    /// 获取索引统计信息
    pub fn get_index_statistics(&self, index_name: &str) -> Option<IndexStatistics> {
        self.index_stats.read().get(index_name).cloned()
    }

    /// 估算值的大小（字节）
    fn estimate_value_size(&self, value: &Value) -> usize {
        match value {
            Value::Null => 1,
            Value::Boolean(_) => 1,
            Value::Integer(_) => 8,
            Value::Real(_) => 8,
            Value::Text(s) => s.len() + 8, // 字符串长度 + 头部
            Value::Timestamp(_) => 16, // 时间戳大小
            Value::Decimal(_) => 16, // 小数大小
        }
    }

    /// 比较两个值的大小
    fn compare_values(&self, a: &Value, b: &Value) -> i32 {
        match (a, b) {
            (Value::Integer(a), Value::Integer(b)) => a.cmp(b) as i32,
            (Value::Real(a), Value::Real(b)) => a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal) as i32,
            (Value::Text(a), Value::Text(b)) => a.cmp(b) as i32,
            (Value::Boolean(a), Value::Boolean(b)) => a.cmp(b) as i32,
            _ => 0, // 不同类型认为相等
        }
    }

    /// 检查统计信息是否过期
    pub fn is_statistics_stale(&self, table_name: &str) -> bool {
        if let Some(stats) = self.get_table_statistics(table_name) {
            let age = SystemTime::now()
                .duration_since(stats.last_updated)
                .unwrap_or_default()
                .as_secs();
            age > self.config.max_age_seconds
        } else {
            true // 没有统计信息认为过期
        }
    }

    /// 更新统计信息配置
    pub fn update_config(&mut self, config: StatisticsConfig) {
        self.config = config;
        tracing::debug!("统计信息配置已更新");
    }
}

impl Default for StatisticsConfig {
    fn default() -> Self {
        Self {
            max_age_seconds: 3600, // 1小时
            histogram_buckets: 100,
            auto_update_threshold: 0.1, // 10%变化
            sampling_rate: 0.1, // 10%采样
        }
    }
}

/// 成本估算模型
/// 基于统计信息估算查询执行计划的成本
#[derive(Debug)]
pub struct CostModel {
    /// 统计信息收集器
    statistics: Arc<StatisticsCollector>,
    /// 成本参数
    cost_params: CostParameters,
}

/// 成本参数
/// 定义各种操作的相对成本
#[derive(Debug, Clone)]
pub struct CostParameters {
    /// 顺序扫描每行的成本
    pub seq_scan_cost_per_row: f64,
    /// 索引扫描每行的成本
    pub index_scan_cost_per_row: f64,
    /// 随机页面访问成本
    pub random_page_cost: f64,
    /// 顺序页面访问成本
    pub sequential_page_cost: f64,
    /// CPU处理每行的成本
    pub cpu_cost_per_row: f64,
    /// 内存操作成本
    pub memory_cost_per_byte: f64,
    /// JOIN操作基础成本
    pub join_base_cost: f64,
    /// 排序操作基础成本
    pub sort_base_cost: f64,
}

impl Default for CostParameters {
    fn default() -> Self {
        Self {
            seq_scan_cost_per_row: 1.0,
            index_scan_cost_per_row: 0.1,
            random_page_cost: 4.0,
            sequential_page_cost: 1.0,
            cpu_cost_per_row: 0.01,
            memory_cost_per_byte: 0.000001,
            join_base_cost: 10.0,
            sort_base_cost: 5.0,
        }
    }
}

impl CostModel {
    /// 创建新的成本模型
    pub fn new(statistics: Arc<StatisticsCollector>) -> Self {
        Self {
            statistics,
            cost_params: CostParameters::default(),
        }
    }

    /// 使用自定义参数创建成本模型
    pub fn with_parameters(statistics: Arc<StatisticsCollector>, params: CostParameters) -> Self {
        Self {
            statistics,
            cost_params: params,
        }
    }

    /// 估算逻辑计划的执行成本
    pub fn estimate_cost(&self, plan: &LogicalPlan) -> Result<f64> {
        match plan {
            LogicalPlan::TableScan { table_name, .. } => {
                self.estimate_table_scan_cost(table_name)
            }
            LogicalPlan::Filter { input, predicate } => {
                let input_cost = self.estimate_cost(input)?;
                let filter_cost = self.estimate_filter_cost(input, predicate)?;
                Ok(input_cost + filter_cost)
            }
            LogicalPlan::Projection { input, .. } => {
                let input_cost = self.estimate_cost(input)?;
                let projection_cost = self.estimate_projection_cost(input)?;
                Ok(input_cost + projection_cost)
            }
            LogicalPlan::Join { left, right, join_type, .. } => {
                let left_cost = self.estimate_cost(left)?;
                let right_cost = self.estimate_cost(right)?;
                let join_cost = self.estimate_join_cost(left, right, join_type)?;
                Ok(left_cost + right_cost + join_cost)
            }
            LogicalPlan::Aggregate { input, .. } => {
                let input_cost = self.estimate_cost(input)?;
                let aggregate_cost = self.estimate_aggregate_cost(input)?;
                Ok(input_cost + aggregate_cost)
            }
            LogicalPlan::Sort { input, .. } => {
                let input_cost = self.estimate_cost(input)?;
                let sort_cost = self.estimate_sort_cost(input)?;
                Ok(input_cost + sort_cost)
            }
            LogicalPlan::Limit { input, .. } => {
                // Limit操作成本很低，主要是输入成本
                self.estimate_cost(input)
            }
            LogicalPlan::CreateTable { .. } => {
                // CREATE TABLE是DDL操作，成本固定且较低
                Ok(1.0)
            }
            LogicalPlan::Insert { values, .. } => {
                // INSERT成本主要取决于插入的行数
                Ok(values.len() as f64 * self.cost_params.cpu_cost_per_row)
            }
            LogicalPlan::Update { .. } => {
                // UPDATE成本固定，后续可以根据条件和更新行数优化
                Ok(10.0 * self.cost_params.cpu_cost_per_row)
            }
            LogicalPlan::Delete { .. } => {
                // DELETE成本固定，后续可以根据条件和删除行数优化
                Ok(5.0 * self.cost_params.cpu_cost_per_row)
            }
        }
    }

    /// 估算表扫描成本
    fn estimate_table_scan_cost(&self, table_name: &str) -> Result<f64> {
        if let Some(stats) = self.statistics.get_table_statistics(table_name) {
            let row_cost = stats.row_count as f64 * self.cost_params.seq_scan_cost_per_row;
            let page_cost = stats.page_count as f64 * self.cost_params.sequential_page_cost;
            Ok(row_cost + page_cost)
        } else {
            // 没有统计信息时使用默认估算
            tracing::warn!("表 {} 没有统计信息，使用默认成本估算", table_name);
            Ok(1000.0) // 默认成本
        }
    }

    /// 估算过滤操作成本
    fn estimate_filter_cost(&self, input: &LogicalPlan, _predicate: &Expression) -> Result<f64> {
        let input_rows = self.estimate_output_rows(input)?;
        // 过滤操作主要是CPU成本
        Ok(input_rows * self.cost_params.cpu_cost_per_row)
    }

    /// 估算投影操作成本
    fn estimate_projection_cost(&self, input: &LogicalPlan) -> Result<f64> {
        let input_rows = self.estimate_output_rows(input)?;
        // 投影操作主要是CPU成本
        Ok(input_rows * self.cost_params.cpu_cost_per_row * 0.5)
    }

    /// 估算JOIN操作成本
    fn estimate_join_cost(&self, left: &LogicalPlan, right: &LogicalPlan,
                         _join_type: &JoinType) -> Result<f64> {
        let left_rows = self.estimate_output_rows(left)?;
        let right_rows = self.estimate_output_rows(right)?;

        // 使用Hash Join的成本模型
        let build_cost = right_rows * self.cost_params.cpu_cost_per_row;
        let probe_cost = left_rows * self.cost_params.cpu_cost_per_row;
        let memory_cost = right_rows * 100.0 * self.cost_params.memory_cost_per_byte; // 假设每行100字节

        Ok(self.cost_params.join_base_cost + build_cost + probe_cost + memory_cost)
    }

    /// 估算聚合操作成本
    fn estimate_aggregate_cost(&self, input: &LogicalPlan) -> Result<f64> {
        let input_rows = self.estimate_output_rows(input)?;
        // 聚合操作需要处理所有输入行
        Ok(input_rows * self.cost_params.cpu_cost_per_row * 2.0)
    }

    /// 估算排序操作成本
    fn estimate_sort_cost(&self, input: &LogicalPlan) -> Result<f64> {
        let input_rows = self.estimate_output_rows(input)?;
        // 排序成本是 O(n log n)
        let sort_factor = if input_rows > 0.0 {
            input_rows * input_rows.log2()
        } else {
            0.0
        };
        Ok(self.cost_params.sort_base_cost + sort_factor * self.cost_params.cpu_cost_per_row)
    }

    /// 估算计划输出的行数
    fn estimate_output_rows(&self, plan: &LogicalPlan) -> Result<f64> {
        match plan {
            LogicalPlan::TableScan { table_name, .. } => {
                if let Some(stats) = self.statistics.get_table_statistics(table_name) {
                    Ok(stats.row_count as f64)
                } else {
                    Ok(1000.0) // 默认估算
                }
            }
            LogicalPlan::Filter { input, .. } => {
                let input_rows = self.estimate_output_rows(input)?;
                // 假设过滤条件的选择性为30%
                Ok(input_rows * 0.3)
            }
            LogicalPlan::Projection { input, .. } => {
                // 投影不改变行数
                self.estimate_output_rows(input)
            }
            LogicalPlan::Join { left, right, .. } => {
                let left_rows = self.estimate_output_rows(left)?;
                let right_rows = self.estimate_output_rows(right)?;
                // 简化的JOIN选择性估算
                Ok(left_rows * right_rows * 0.1)
            }
            LogicalPlan::Aggregate { input, group_expressions, .. } => {
                if group_expressions.is_empty() {
                    // 没有GROUP BY，输出一行
                    Ok(1.0)
                } else {
                    let input_rows = self.estimate_output_rows(input)?;
                    // 假设GROUP BY的选择性为10%
                    Ok(input_rows * 0.1)
                }
            }
            LogicalPlan::Sort { input, .. } => {
                // 排序不改变行数
                self.estimate_output_rows(input)
            }
            LogicalPlan::Limit { count, .. } => {
                Ok(*count as f64)
            }
            LogicalPlan::CreateTable { .. } => {
                // CREATE TABLE不产生输出行
                Ok(0.0)
            }
            LogicalPlan::Insert { .. } => {
                // INSERT不产生输出行
                Ok(0.0)
            }
            LogicalPlan::Update { .. } => {
                // UPDATE不产生输出行
                Ok(0.0)
            }
            LogicalPlan::Delete { .. } => {
                // DELETE不产生输出行
                Ok(0.0)
            }
        }
    }

    /// 更新成本参数
    pub fn update_parameters(&mut self, params: CostParameters) {
        self.cost_params = params;
        tracing::debug!("成本模型参数已更新");
    }

    /// 获取当前成本参数
    pub fn get_parameters(&self) -> &CostParameters {
        &self.cost_params
    }
}

/// 计划缓存
/// 缓存已优化的查询计划，避免重复优化
#[derive(Debug)]
pub struct PlanCache {
    /// 缓存的计划
    cache: RwLock<HashMap<String, CachedPlan>>,
    /// 缓存配置
    config: PlanCacheConfig,
    /// 缓存统计信息
    stats: RwLock<PlanCacheStats>,
}

/// 缓存的计划
#[derive(Debug, Clone)]
pub struct CachedPlan {
    /// 优化后的计划
    pub plan: LogicalPlan,
    /// 估算成本
    pub estimated_cost: f64,
    /// 创建时间
    pub created_at: SystemTime,
    /// 访问次数
    pub access_count: u64,
    /// 最后访问时间
    pub last_accessed: SystemTime,
}

/// 计划缓存配置
#[derive(Debug, Clone)]
pub struct PlanCacheConfig {
    /// 最大缓存条目数
    pub max_entries: usize,
    /// 缓存过期时间（秒）
    pub ttl_seconds: u64,
    /// 是否启用缓存
    pub enabled: bool,
}

/// 计划缓存统计信息
#[derive(Debug, Clone, Default)]
pub struct PlanCacheStats {
    /// 缓存命中次数
    pub hits: u64,
    /// 缓存未命中次数
    pub misses: u64,
    /// 缓存驱逐次数
    pub evictions: u64,
    /// 当前缓存条目数
    pub current_entries: usize,
}

impl PlanCache {
    /// 创建新的计划缓存
    pub fn new() -> Self {
        Self {
            cache: RwLock::new(HashMap::new()),
            config: PlanCacheConfig::default(),
            stats: RwLock::new(PlanCacheStats::default()),
        }
    }

    /// 使用自定义配置创建计划缓存
    pub fn with_config(config: PlanCacheConfig) -> Self {
        Self {
            cache: RwLock::new(HashMap::new()),
            config,
            stats: RwLock::new(PlanCacheStats::default()),
        }
    }

    /// 获取缓存的计划
    pub fn get(&self, query_hash: &str) -> Option<LogicalPlan> {
        if !self.config.enabled {
            return None;
        }

        let mut cache = self.cache.write();
        let mut stats = self.stats.write();

        if let Some(cached) = cache.get_mut(query_hash) {
            // 检查是否过期
            let age = SystemTime::now()
                .duration_since(cached.created_at)
                .unwrap_or_default()
                .as_secs();

            if age <= self.config.ttl_seconds {
                // 更新访问信息
                cached.access_count += 1;
                cached.last_accessed = SystemTime::now();

                stats.hits += 1;
                tracing::debug!("计划缓存命中: {}", query_hash);
                return Some(cached.plan.clone());
            } else {
                // 过期，移除
                cache.remove(query_hash);
                stats.evictions += 1;
                tracing::debug!("计划缓存过期: {}", query_hash);
            }
        }

        stats.misses += 1;
        tracing::debug!("计划缓存未命中: {}", query_hash);
        None
    }

    /// 缓存计划
    pub fn put(&self, query_hash: String, plan: LogicalPlan, cost: f64) {
        if !self.config.enabled {
            return;
        }

        let mut cache = self.cache.write();
        let mut stats = self.stats.write();

        // 检查缓存大小限制
        if cache.len() >= self.config.max_entries {
            // 使用LRU策略驱逐最久未访问的条目
            if let Some(lru_key) = self.find_lru_key(&cache) {
                cache.remove(&lru_key);
                stats.evictions += 1;
                tracing::debug!("驱逐LRU计划: {}", lru_key);
            }
        }

        let cached_plan = CachedPlan {
            plan,
            estimated_cost: cost,
            created_at: SystemTime::now(),
            access_count: 0,
            last_accessed: SystemTime::now(),
        };

        cache.insert(query_hash.clone(), cached_plan);
        stats.current_entries = cache.len();

        tracing::debug!("缓存新计划: {}, 成本: {:.2}", query_hash, cost);
    }

    /// 查找最久未访问的键
    fn find_lru_key(&self, cache: &HashMap<String, CachedPlan>) -> Option<String> {
        cache.iter()
            .min_by_key(|(_, plan)| plan.last_accessed)
            .map(|(key, _)| key.clone())
    }

    /// 清空缓存
    pub fn clear(&self) {
        let mut cache = self.cache.write();
        let mut stats = self.stats.write();

        cache.clear();
        stats.current_entries = 0;
        stats.evictions += cache.len() as u64;

        tracing::debug!("计划缓存已清空");
    }

    /// 获取缓存统计信息
    pub fn get_stats(&self) -> PlanCacheStats {
        self.stats.read().clone()
    }

    /// 获取缓存命中率
    pub fn hit_rate(&self) -> f64 {
        let stats = self.stats.read();
        let total = stats.hits + stats.misses;
        if total > 0 {
            stats.hits as f64 / total as f64
        } else {
            0.0
        }
    }

    /// 更新缓存配置
    pub fn update_config(&mut self, config: PlanCacheConfig) {
        self.config = config;
        tracing::debug!("计划缓存配置已更新");
    }
}

impl Default for PlanCacheConfig {
    fn default() -> Self {
        Self {
            max_entries: 1000,
            ttl_seconds: 3600, // 1小时
            enabled: true,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sql::planner::{Expression, LiteralValue};
    
    #[test]
    fn test_optimizer_creation() {
        let optimizer = QueryOptimizer::new();
        assert_eq!(optimizer.rules.len(), 5); // 现在有5个优化规则
        assert_eq!(optimizer.stats.optimization_count, 0);
    }
    
    #[test]
    fn test_predicate_pushdown() {
        let rule = PredicatePushdownRule;
        assert_eq!(rule.name(), "PredicatePushdown");
        
        // 创建简单的测试计划
        let table_scan = LogicalPlan::TableScan {
            table_name: "users".to_string(),
            alias: None,
            projection: None,
        };
        
        let filter = LogicalPlan::Filter {
            input: Box::new(table_scan),
            predicate: Expression::Literal(LiteralValue::Boolean(true)),
        };
        
        let result = rule.optimize(filter);
        assert!(result.is_ok());
    }
    
    #[test]
    fn test_optimization_rule_names() {
        let predicate_rule = PredicatePushdownRule;
        let projection_rule = ProjectionPushdownRule;
        let constant_rule = ConstantFoldingRule;
        let join_rule = JoinReorderRule;
        let index_rule = IndexSelectionRule;

        assert_eq!(predicate_rule.name(), "PredicatePushdown");
        assert_eq!(projection_rule.name(), "ProjectionPushdown");
        assert_eq!(constant_rule.name(), "ConstantFolding");
        assert_eq!(join_rule.name(), "JoinReorder");
        assert_eq!(index_rule.name(), "IndexSelection");
    }

    #[test]
    fn test_optimizer_stats() {
        let mut optimizer = QueryOptimizer::new();
        let stats = optimizer.get_stats();
        assert_eq!(stats.optimization_count, 0);
        assert_eq!(stats.rule_applications, 0);

        // 重置统计信息
        optimizer.reset_stats();
        let stats = optimizer.get_stats();
        assert_eq!(stats.optimization_count, 0);
    }

    #[test]
    fn test_join_reorder_rule() {
        let rule = JoinReorderRule;
        assert_eq!(rule.name(), "JoinReorder");

        // 测试成本估算
        let cost = rule.estimate_plan_cost(&LogicalPlan::TableScan {
            table_name: "test".to_string(),
            alias: None,
            projection: None,
        });
        assert_eq!(cost, 100);
    }

    #[test]
    fn test_index_selection_rule() {
        let rule = IndexSelectionRule;
        assert_eq!(rule.name(), "IndexSelection");

        // 测试索引选择
        let can_use = rule.can_use_index("test_table", &Expression::Literal(LiteralValue::Integer(1)));
        assert_eq!(can_use, false); // 简化实现总是返回false
    }
}
